package mqtt

import (
	"errors"
	"fmt"
	"lifi"
	"lifi/plugins/common/mqtt"
	"lifi/plugins/sinks"
	"log"
	"sync"
	"time"
)

type message struct {
	topic   string
	payload []byte
}

type MQTT struct {
	Topic string `yaml:"topic" json:"topic,omitempty"`
	mqtt.MqttConfig

	client mqtt.Client

	sync.Mutex
	serializer lifi.Serializer
}

func (m *MQTT) Init() error {
	if len(m.Servers) == 0 {
		return errors.New("no servers specified")
	}

	if m.PersistentSession && m.ClientID == "" {
		return errors.New("persistent_session requires client_id")
	}
	if m.QoS > 2 || m.QoS < 0 {
		return fmt.Errorf("qos value must be 0, 1, or 2: %d", m.QoS)
	}

	return nil
}

func (m *MQTT) Connect() error {
	m.Lock()
	defer m.Unlock()

	client, err := mqtt.NewClient(&m.MqttConfig)
	if err != nil {
		return err
	}
	m.client = client

	_, err = m.client.Connect()
	return err
}

func (m *MQTT) SetSerializer(serializer lifi.Serializer) {
	m.serializer = serializer
}

func (m *MQTT) Close() error {
	// Unregister devices if Homie layout was used. Usually we should do this
	// using a "will" message, but this can only be done at connect time where,
	// due to the dynamic nature of Telegraf messages, we do not know the topics
	// to issue that "will" yet.

	return m.client.Close()
}

func (m *MQTT) Write(data ...lifi.Data) error {
	m.Lock()
	defer m.Unlock()
	if len(data) == 0 {
		return nil
	}

	// Group the metrics to topics and serialize them

	topicMessages := m.convertDataToMessage(data)

	for _, msg := range topicMessages {
		if err := m.client.Publish(msg.topic, msg.payload); err != nil {
			log.Printf("Could not publish message to MQTT server: %v", err)
		}
	}

	return nil
}

func (m *MQTT) convertDataToMessage(dataList []lifi.Data) []message {
	collection := make([]message, 0, len(dataList))
	for _, metric := range dataList {

		buf, err := m.serializer.Serialize(metric)
		if err != nil {
			log.Printf("Could not serialize metric for topic %q: %v", m.Topic, err)
			log.Printf("metric was: %v", metric)
			continue
		}
		collection = append(collection, message{m.Topic, buf})
	}

	return collection
}

func init() {
	sinks.Add("mqtt", func() lifi.Sink {
		return &MQTT{
			MqttConfig: mqtt.MqttConfig{
				KeepAlive:     30,
				Timeout:       time.Duration(5 * time.Second),
				AutoReconnect: true,
			},
		}
	})
}
